package com.bujunjie.study.realtime.dim.function;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import com.bujunjie.study.realtime.common.enums.FlinkOpEnum;
import com.bujunjie.study.realtime.common.util.JdbcUtil;
import com.bujunjie.study.realtime.dim.model.vo.DimTableVO;
import com.bujunjie.study.realtime.dim.model.vo.Mysql2MaxwellVO;
import com.bujunjie.study.realtime.dim.model.vo.ProductVO;
import com.google.common.collect.Maps;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * <p>双流join </p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/13 9:58
 */
public class TableProcessFunction extends BroadcastProcessFunction<Mysql2MaxwellVO<ProductVO>, DimTableVO, Tuple2<Map<String, Object>, DimTableVO>> {

    /**
     * 描述广播状态
     */
    private MapStateDescriptor<String, DimTableVO> mapStateDescriptor;

    /**
     * 项目处理化的配置信息
     */
    private Map<String, DimTableVO> configMap = Maps.newHashMap();

    public TableProcessFunction(MapStateDescriptor<String, DimTableVO> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        //项目初始化从数据库读取配置信息，防止刚开始没有配置信息
        Connection mySQLConnection = JdbcUtil.getMySQLConnection();
        List<DimTableVO> tableProcessDimList = JdbcUtil.queryList(mySQLConnection, "select * from gmall2024_config.table_process_dim", DimTableVO.class, true);
        for (DimTableVO dimTableVO : tableProcessDimList) {
            configMap.put(dimTableVO.getSourceTable(), dimTableVO);
        }
        JdbcUtil.closeMySQLConnection(mySQLConnection);
    }


    @Override
    public void processElement(Mysql2MaxwellVO<ProductVO> maxwellVO, BroadcastProcessFunction<Mysql2MaxwellVO<ProductVO>, DimTableVO,
            Tuple2<Map<String, Object>, DimTableVO>>.ReadOnlyContext ctx, Collector<Tuple2<Map<String, Object>, DimTableVO>> out) throws Exception {
        String table = maxwellVO.getTable();
        ReadOnlyBroadcastState<String, DimTableVO> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if (broadcastState.contains(table) || configMap.containsKey(table)) {
            //  代表是配置中的表，维度数据
            DimTableVO dimTableVO = this.getDimTableVO(table, broadcastState);
            ProductVO data = maxwellVO.getData();
            String sinkColumns = dimTableVO.getSinkColumns();
            // 过滤出复合条件的字段值
            Map<String, Object> filterMap = this.filterNeedColumns(data, sinkColumns);
            filterMap.put("type", data.getType());
            out.collect(Tuple2.of(filterMap, dimTableVO));
        }
    }


    /**
     * <p>获取维度表数据</p>
     *
     * @param table          表名
     * @param broadcastState 广播状态数据
     * @return 返回对应的维度表数据
     * @author bu.junjie
     * @date 2025/9/13 10:41
     */
    private DimTableVO getDimTableVO(String table, ReadOnlyBroadcastState<String, DimTableVO> broadcastState) throws Exception {
        if (broadcastState.contains(table)) {
            return broadcastState.get(table);
        }
        if (configMap.containsKey(table)) {
            return configMap.get(table);
        }
        return null;
    }

    @Override
    public void processBroadcastElement(DimTableVO dimTable, BroadcastProcessFunction<Mysql2MaxwellVO<ProductVO>, DimTableVO,
            Tuple2<Map<String, Object>, DimTableVO>>.Context ctx, Collector<Tuple2<Map<String, Object>, DimTableVO>> out) throws Exception {
        // 获取配置表的操作类型
        String op = dimTable.getOp();
        //  获取广播状态
        BroadcastState<String, DimTableVO> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        // 获取维度表的名称
        String sourceTable = dimTable.getSourceTable();
        if (FlinkOpEnum.D.getOp().equals(op)) {
            broadcastState.remove(sourceTable);
            configMap.remove(sourceTable);
        } else {
            // 往原数据里面添加维度表数据
            broadcastState.put(sourceTable, dimTable);
            configMap.put(sourceTable, dimTable);
        }
    }


    /**
     * <p>
     * 过滤出需要的字段
     *     <ol>
     *         <li>{"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"555","id":1}</li>
     *         <li>sinkColumns  id,tm_name</li>
     *     </ol>
     * </p>
     *
     * @param
     * @return 返回复合规范的字段
     * @author bu.junjie
     * @date 2025/9/15 8:54
     */
    private Map<String, Object> filterNeedColumns(ProductVO productVO, String sinkColumns) {
        Map<String, Object> filterMap = Maps.newHashMap();
        List<String> columnList = Arrays.asList(sinkColumns.split(StrUtil.COMMA));
        Map<String, Object> propMap = BeanUtil.beanToMap(productVO, true, false);
        propMap.forEach((key, value) -> {
            if (columnList.contains(key)) {
                filterMap.put(key, value);
            }
        });
        return filterMap;
    }


}
